package com.tg.rxjava;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func1;
import rx.observables.BlockingObservable;
import rx.schedulers.Schedulers;

public class RxJavaAsyncTest {

    public static void main(String[] args) throws Exception {
        BlockingObservable<Long> a;

        Observable<Long> observable = Observable.just(1, 2)
                .subscribeOn(Schedulers.io())
                .map(t -> {
                    try {
                        Thread.sleep(1000); //耗时的操作
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return (long) (t * 2);

                });

        observable.subscribe(getSubscriber("a"));
        observable.subscribe(getSubscriber("b"));
        observable.subscribe(getSubscriber("c"));

        System.out.println("other...");
        Thread.sleep(5000); //耗时的操作
    }

    private static Subscriber<Long> getSubscriber(String name) {
        return new Subscriber<Long>() {
            @Override
            public void onCompleted() {
                System.out.println(name + " onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("error" + e);
            }

            @Override
            public void onNext(Long result) {
                System.out.println(Thread.currentThread().getName() + "-result-" + name + "-" + result);
            }
        };
    }
}
